package com.qingcloud.df.executor.cluster;

import com.qingcloud.base.remote.executor.ExecutorParam;
import com.qingcloud.df.executor.worker.ExecutorService;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.context.ConfigManager;
import com.qingcloud.df.sdk.component.message.NodeInfo;
import com.qingcloud.df.sdk.component.message.request.*;
import com.qingcloud.df.sdk.component.message.response.ExecutorStatus;
import com.qingcloud.df.sdk.component.message.response.JobResponse;
import com.qingcloud.df.sdk.component.transport.handler.ServerHandlerUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
public class JobService {
    public static final Logger logger = LoggerFactory.getLogger(ExecutorService.class);

    /**
     * 未分配Job 找合适的执行器，并分配给该执行器
     * @param jobParams
     * @return
     * @throws Exception
     */
    public  void assignJobs(List<ExecutorParam.JobParam> jobParams) {
        if(jobParams!=null) {
            for (ExecutorParam.JobParam jobParam : jobParams) {
                assignJob(jobParam);
            }
        }
    }

    /**
     * 分配Job到执行器节点
     * @param jobParam
     */
    public  void assignJob(ExecutorParam.JobParam jobParam)  {
        Integer jobId  =  jobParam.getJobId();
        String env = jobParam.getEnv();
        List<CompInstContext> list  = ConfigManager.loadByJobId(jobParam.getJobId(),jobParam.getEnv());
        assignJob(jobId,env,list);
    }

    /**
     * 分配Job到执行器节点
     * @param jobId
     * @param env
     * @param list
     */
    public  void assignJob(Integer jobId, String env, List<CompInstContext> list)  {
        JobInstanceRequest jobInstanceRequest = new JobInstanceRequest();
        jobInstanceRequest.setEnv(env);
        jobInstanceRequest.setJobId(jobId);
        jobInstanceRequest.setCompInstList(list);
        NodeInfo nodeInfo =  ClusterManager.selectAvailableNode(env);
        if(nodeInfo == null) {
            logger.error("have no suitable executor node, please check!");
            return ;
        }
        String nodeId = nodeInfo.getAkkaAddress();
        ServerHandlerUtil.assiginJob(nodeId,jobInstanceRequest);
        logger.info("job [{}] has assign to node [{}]! please waiting response message!",jobId,nodeId);
    }

    /**
     * 分配Instance实例到执行器节点
     * @param instanceParam
     * @throws Exception
     */
    public  void assignInstance(ExecutorParam.InstanceParam instanceParam)  {
        CompInstContext compInstContext = ConfigManager.loadByInstanceId(instanceParam.getInstanceId(),instanceParam.getEnv());
        NodeInfo nodeInfo =  ClusterManager.selectAvailableNode(instanceParam.getEnv());
        if(nodeInfo == null) {
            logger.error("have no suitable executor node, please check!");
            return ;
        }
        String nodeId = nodeInfo.getAkkaAddress();
        ServerHandlerUtil.assiginInstance(nodeId,compInstContext);
    }

    /**
     * 分配Instance 到指定的Node
     * @param instanceParam
     * @param nodeId
     */
    public  void assignInstance(ExecutorParam.InstanceParam instanceParam,String nodeId)  {
        CompInstContext compInstContext = ConfigManager.loadByInstanceId(instanceParam.getInstanceId(),instanceParam.getEnv());
        ServerHandlerUtil.assiginInstance(nodeId,compInstContext);
    }

    /**
     * 删除job
     * @param nodeAddress
     * @param request
     * @return
     */
    public  List<ExecutorStatus> removeJob(String nodeAddress , JobRemoveRequest request)  {
        JobResponse response =ServerHandlerUtil.removeJob(nodeAddress,request);
        List<ExecutorStatus>  status = response.getStatusList();
        String nodeId = ClusterManager.getNodeByJobId(response.getJobId());
        ClusterManager.removeNodeJob(nodeId,request.getJobId());
        return status;
    }

    /**
     * 启动Job
     * @param nodeAddress
     * @param request
     * @return
     */
    public  List<ExecutorStatus> startJob(String nodeAddress , JobStartRequest request)  {
        JobResponse response =ServerHandlerUtil.startJob(nodeAddress,request);
        return buildResult(response);
    }

    /**
     * 停止Job
     * @param nodeAddress
     * @param request
     * @return
     */
    public  List<ExecutorStatus> stopJob(String nodeAddress , JobStopRequest request)  {
        JobResponse response =ServerHandlerUtil.stopJob(nodeAddress,request);
        return buildResult(response);
    }

    /**
     * 启动实例
     * @param nodeAddress
     * @param request
     * @return
     */
    public  List<ExecutorStatus> startInstance(String nodeAddress , InstanceStartRequest request)  {
        JobResponse response =ServerHandlerUtil.startInstance(nodeAddress,request);
        return buildResult(response);
    }

    /**
     * 停止实例
     * @param nodeAddress
     * @param request
     * @return
     */
    public  List<ExecutorStatus> stopInstance(String nodeAddress , InstanceStopRequest request)  {
        JobResponse response =ServerHandlerUtil.stopInstance(nodeAddress,request);
        return buildResult(response);
    }

    /**
     * 删除实例
     * @param nodeAddress
     * @param request
     * @return
     */
    public  List<ExecutorStatus> removeInstance(String nodeAddress , InstanceRemoveRequest request)  {
        JobResponse response =ServerHandlerUtil.removeInstance(nodeAddress,request);
        List<ExecutorStatus>  status = response.getStatusList();
        String nodeId = ClusterManager.getNodeByJobId(response.getJobId());
        ClusterManager.removeNodeInstance(nodeId,response.getJobId(),request.getInstanceId());
        return status;
    }

    /**
     * 获取返回值并更新cluster状态数据
     * @param response
     * @return
     */
    private  List<ExecutorStatus> buildResult(JobResponse response){
        List<CompInstContext> compInsts = response.getContextMap();
        String nodeId = ClusterManager.getNodeByJobId(response.getJobId());
        ClusterManager.pubNodeJob(nodeId,response.getJobId(),compInsts);
        return response.getStatusList();
    }

}
